#include "P_Queue.h"
#include "I_Thread.h"


BlockNotifyQueue::BlockNotifyQueue()
{
	ink_mutex_init(&lock);
	ink_cond_init(&might_have_data);
}

void
BlockNotifyQueue::enqueue(Event *e, bool fast_signal)
{
	e->in_queue = 1;
	//任务执行者,也就是当前队列的持有者
	EThread *e_ethread = e->ethread;
	ink_mutex_lock(&lock);
	bool empty = g_que.empty();
	g_que.push(e);
	ink_mutex_release(&lock);
	if(empty) {
		//任务发起者
		EThread *inserting_thread = this_ethread();
		if(e_ethread != inserting_thread) {
			if(!inserting_thread || !inserting_thread->ethreads_tobe_signalled) {
				signal();
				if(fast_signal) {
					if(e_ethread->signal_hook)
						e->ethread->signal_hook(e_ethread);
				}
			}else{
				if(fast_signal) {
					if(e_ethread->signal_hook)
						e->ethread->signal_hook(e_ethread);
				}

				/*if(e_ethread->notifyQueue.try_signal())
					return;
				*/
				//任务发起者,正在执行的当前线程,操作的是自己的线程所属对象
				//所以操作ethreads_tobe_signalled不需要上锁.
				if(inserting_thread->n_ethreads_tobe_signalled > MAX_EVENT_THREAD) {
					signal();
					return;
				}
							
				inserting_thread->ethreads_tobe_signalled[inserting_thread->n_ethreads_tobe_signalled++] = e_ethread;
			}
		}
	}

}

void
BlockNotifyQueue::signal()
{
	ink_mutex_lock(&lock);
	ink_cond_signal(&might_have_data);
	ink_mutex_release(&lock);
}

bool
BlockNotifyQueue::try_signal()
{
	if(ink_mutex_try_lock(&lock)) {
		ink_cond_signal(&might_have_data);
		ink_mutex_release(&lock);
		return true;
	}else{
		return false;
	}
}

void
BlockNotifyQueue::enqueue_local_all()
{
	std::queue<Event *> tmp_que;
	ink_mutex_lock(&lock);
	g_que.swap(tmp_que);
	ink_mutex_release(&lock);

	while(!tmp_que.empty())
	{
		local_que.push(tmp_que.front());
		tmp_que.pop();	
	}
}

void
BlockNotifyQueue::enqueue_local(Event *e)
{
	e->in_queue = 1;
	local_que.push(e);
}

Event *
BlockNotifyQueue::dequeue_local()
{
	if(local_que.empty())
		return NULL;

	Event *e = local_que.front();
	local_que.pop();
	e->in_queue = 0;
	return e;
}

void
BlockNotifyQueue::dequeue_time(ink_hrtime cur_time, ink_hrtime timeout, bool sleep)
{
	(void)cur_time;
	ink_mutex_lock(&lock);
	if(sleep){
		//if(g_que.empty() && local_que.empty()){
		if(g_que.empty()){
			ink_timestruc time_wait = ink_hrtime_to_timespec(timeout);
			ink_cond_time_wait(&might_have_data, &lock, &time_wait);
		}		
	}
	ink_mutex_release(&lock);

	enqueue_local_all();
}

void flush_signal(EThread *thr)
{
	int n = thr->n_ethreads_tobe_signalled;
	for(int i = 0; i < n; i++){
		if(thr->ethreads_tobe_signalled[i])
		{
			if(thr->ethreads_tobe_signalled[i]->signal_hook)
				thr->ethreads_tobe_signalled[i]->signal_hook(thr->ethreads_tobe_signalled[i]);
			thr->ethreads_tobe_signalled[i]->notifyQueue.signal();
			thr->ethreads_tobe_signalled[i] = NULL;
		}
	}
	thr->n_ethreads_tobe_signalled = 0;	
}

void
PriorityQueue::enqueue(Event *e, ink_hrtime time)
{
	e->in_queue = 1;
	PriorityEvent pe;
	pe.e = e;
	pe.exec_time = time + e->timeout_at;

	prior_list.insert(prior_list.end(), pe);
}

void
PriorityQueue::check_ready(ink_hrtime time)
{
	prior_list.sort();
	std::list<PriorityEvent>::iterator it;

	for(it = prior_list.begin(); it != prior_list.end();){
		if((*it).exec_time < time + ink_hrtime_from_msecond(5)){
			exec_list.push_back(*it);
			it = prior_list.erase(it);
		}else
			break;
	}
}

ink_hrtime
PriorityQueue::get_near_exec_time()
{
	if(prior_list.empty())
		return 0;
	PriorityEvent pe = prior_list.front();
	return pe.exec_time;
}

Event *
PriorityQueue::dequeue()
{
	if(!exec_list.empty())
	{
		PriorityEvent pe = exec_list.front();
		exec_list.pop_front();
		pe.e->in_queue = 0;
		return pe.e;
	}

	return NULL;
}


void
NegativeQueue::enqueue(Event *e)
{
	e->in_queue = 1;
	std::list<Event *>::iterator it, pre_it;

	for(pre_it = it = neg_list.begin(); it != neg_list.end(); it++) {
		pre_it = it;
		if(e->timeout_at > (*it)->timeout_at)
			break;
	}

	if(it == neg_list.end())
		pre_it = it;
	neg_list.insert(pre_it, e);
}

Event *
NegativeQueue::dequeue()
{
	if(neg_list.empty())
		return NULL;

	Event *e = neg_list.front();
	neg_list.pop_front();
	e->in_queue = 0;
	return e;
}

bool
NegativeQueue::empty()
{
	return neg_list.empty();
}
